package docker

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"github.com/docker/docker/api/types"
	"github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/mount"
	"github.com/docker/docker/client"
	"github.com/docker/docker/pkg/stdcopy"
	"github.com/sirupsen/logrus"
	"github.com/tluo-github/super-runner/common"
	"github.com/tluo-github/super-runner/executors"
	docker_helper "github.com/tluo-github/super-runner/helpers/docker"
	"github.com/tluo-github/super-runner/helpers/kafka"
	"github.com/tluo-github/super-runner/helpers/minio"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)
type DockerExecutor struct {
	Build       common.Build
	config 		common.RunnerConfig
	dockerClient *client.Client
	buildContainer *types.ContainerJSON
	buildFinish chan error
	containerName string
	Context context.Context
	BuildLog    *os.File
	IsSystemError bool
	ObjectName  string
	sync.Mutex

	CI_PROJECT_NAME *common.JobVariable
	CI_PROJECT_URL *common.JobVariable
	CI_PROJECT_NAMESPACE	*common.JobVariable
	CI_PROJECT_BRANCH	*common.JobVariable
	CI_BUILD_STATE	*common.JobVariable
	CI_PROJECT_DIR 	*common.JobVariable

}

func (d *DockerExecutor) Prepare(config common.RunnerConfig, build common.Build) error {
	d.dockerClient = docker_helper.GetDockerClient(config.Docker)
	d.config = config
	d.Build = build
	d.buildFinish = make(chan error, 1)
	d.IsSystemError = false

	d.containerName = fmt.Sprintf("super-runner-%s-%d",d.Build.Name,d.Build.Timestamp)

	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Prepare")

	// 发送开始执行消息给 kafka
	kafka.StartKafka(config.KafkaConfig,d.containerName,build.MetaData,config.Name)

	metaData := build.MetaData

	// 创建 build log
	filename := fmt.Sprintf("/logs/%s.log",d.containerName)
	build_log, err := os.Create(filename)
	if err != nil {
		d.IsSystemError = true
		return err
	}
	d.BuildLog = build_log
	timeStr:=time.Now().Format("2006-01-02")
	d.ObjectName = fmt.Sprintf("%s/%s",timeStr,filepath.Base(d.BuildLog.Name()))

	// 从 state 是否为 verify 判断是否是最后一个 job

	// CI_PROJECT_NAME 项目名称
	d.CI_PROJECT_NAME = &common.JobVariable{
		Key:   "CI_PROJECT_NAME",
		Value: fmt.Sprint(metaData["CI_PROJECT_NAME"]),
	}
	//CI_PROJECT_URL 项目 gitlab url 路径
	d.CI_PROJECT_URL =&common.JobVariable{
		Key:   "CI_PROJECT_URL",
		Value: fmt.Sprint(metaData["CI_PROJECT_URL"]),
	}
	//CI_PROJECT_NAMESPACE 项目所属组
	d.CI_PROJECT_NAMESPACE =&common.JobVariable{
		Key:   "CI_PROJECT_NAMESPACE",
		Value: fmt.Sprint(metaData["CI_PROJECT_NAMESPACE"]),
	}
	//CI_PROJECT_BRANCH 分支名称
	d.CI_PROJECT_BRANCH =&common.JobVariable{
		Key:   "CI_PROJECT_BRANCH",
		Value: fmt.Sprint(metaData["CI_PROJECT_BRANCH"]),
	}
	//CI_BUILD_STATE 当前 job 阶段
	d.CI_BUILD_STATE =&common.JobVariable{
		Key:   "CI_BUILD_STATE",
		Value: fmt.Sprint(metaData["CI_BUILD_STATE"]),
	}
	// CI_PROJECT_DIR 项目这次job 的工作目录: /workspace/$PATH_WITH_NAMESPACE/$CI_PROJECT_NAME_d.Build.Timestamp
	d.CI_PROJECT_DIR =&common.JobVariable{
		Key:   "CI_PROJECT_DIR",
		Value: fmt.Sprintf("%s/%s/%s_%d",d.getWorkspace(),d.CI_PROJECT_NAMESPACE.Value,d.CI_PROJECT_NAME.Value,d.Build.Timestamp),
	}
	d.Build.Variables = append(d.Build.Variables,*d.CI_PROJECT_DIR)
	// 将 metaData map 数据全部映射到环境变量
	for key := range metaData{
		logrus.Infof("metaData  : %s = %s",key,metaData[key])
		d.Build.Variables = append(d.Build.Variables, common.JobVariable{
			Key:   key,
			Value: fmt.Sprint(metaData[key]),
		})

	}

	// check 代表第一个 job
	if d.CI_BUILD_STATE.Value == "check" {
		url_parts := strings.Split(d.CI_PROJECT_URL.Value, "://")
		project_url := fmt.Sprintf("%s://%s:%s@%s",url_parts[0], config.GitConfig.Name, config.GitConfig.Password, url_parts[1])
		gitCommand := fmt.Sprintf(" rm -rf %s && git clone %s -b %s %s && cd %s", d.CI_PROJECT_DIR.Value , project_url, d.CI_PROJECT_BRANCH.Value, d.CI_PROJECT_DIR.Value,d.CI_PROJECT_DIR.Value)
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Infoln("gitCommand : ", gitCommand)
		d.Build.Command = fmt.Sprintf("%s && %s",gitCommand, d.Build.Command)
	} else {
		d.Build.Command = fmt.Sprintf("cd %s && %s",d.CI_PROJECT_DIR.Value, d.Build.Command)
	}

	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Command : ", d.Build.Command)

	d.Context = context.Background()

	err = d.getBuildImage()
	if err != nil {
		return err
	}
	return nil
}

// 下载镜像
func (d *DockerExecutor) getBuildImage() error {
	// Fetch image
	reader, err := d.dockerClient.ImagePull(d.Context, d.Build.Image.Name, types.ImagePullOptions{})
	if err != nil {
		return nil
	}
	byte,_ := ioutil.ReadAll(reader)
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("getBuildImage : ", string(byte))

	return nil
}
// 获得所有 VolumeMounts
func (s *DockerExecutor) getVolumeMounts()(mounts []mount.Mount){
	for _, volume := range s.config.Docker.Volumes{
		mounts = append(mounts, mount.Mount{
			Type:          mount.TypeBind,
			Source:        volume.Source,
			Target:        volume.Target,
			ReadOnly:      volume.ReadOnly,
		})
	}
	return
}

func (d *DockerExecutor) createContainer(containerType string, imageDefinition common.Image, allowedInternalImages []string) (*types.ContainerJSON, error) {
	//创建容器
	config := &container.Config{
		Image:        d.Build.Image.Name,		// 镜像名
		Tty:          false,			// docker run命令中的-t选项
		AttachStdin:  true,
		AttachStdout: true,
		AttachStderr: true,
		OpenStdin:    true,				// docker run命令中的-i选项
		StdinOnce:    true,
		Env:          docker_helper.BuildVariables(d.Build.Variables),
		Cmd: []string{"/bin/sh"},
	}
	hostConfig := &container.HostConfig{
		DNS:             d.config.Docker.DNS,
		Privileged: d.config.Docker.Privileged,
		RestartPolicy: container.RestartPolicy{Name: "no"},
		LogConfig: container.LogConfig{
			Type: "json-file",
		},
		Mounts: d.getVolumeMounts(),

	}
	d.Context = context.Background()
	d.removeContainer(d.containerName)


	resp, err := d.dockerClient.ContainerCreate(d.Context,config,hostConfig,nil,d.containerName)
	if err != nil {
		return nil, err
	}
	inspect, err := d.dockerClient.ContainerInspect(d.Context, resp.ID)
	if err != nil {
		return nil, err
	}
	return &inspect, nil
}

func (d *DockerExecutor) requestBuildContainer() (*types.ContainerJSON, error) {
	d.Lock()
	defer d.Unlock()

	if d.buildContainer == nil {
		var err error
		// 启动将运行实际构建的构建容器
		d.buildContainer, err = d.createContainer("build", d.Build.Image, []string{})
		if err != nil {
			return nil, err
		}
	}

	return d.buildContainer, nil
}


func (d* DockerExecutor) runContainer(id string, input io.Reader)  error {
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Starting container ", id )

	err := d.dockerClient.ContainerStart(d.Context, id, types.ContainerStartOptions{})
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorln("Starting container ", id )
		d.IsSystemError = true
		return err
	}
	options := types.ContainerAttachOptions{
		Stream: true,
		Stdin:  true,
		Stdout: true,
		Stderr: true,
	}
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Attaching to  container ", id )

	hijacked, err := d.dockerClient.ContainerAttach(d.Context, id, options)
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorln("Attaching to container faild " )
		d.IsSystemError = true
		return err
	}
	defer hijacked.Close()


	attachCh := make(chan error, 2)

	// 将容器标准输入日志,输出到日志文件
	go func() {
		_, err := stdcopy.StdCopy(d.BuildLog, d.BuildLog, hijacked.Reader)
		if err != nil {
			attachCh <- err
		}
	}()

	// 将输入写入容器并关闭其STDIN以使其完成
	go func() {
		_, err := io.Copy(hijacked.Conn, input)
		hijacked.CloseWrite()
		if err != nil {
			attachCh <- err
		}
	}()

	waitCh := make(chan error, 1)
	go func() {
		waitCh <- d.waitForContainer(d.Context, id)
	}()

	select {
	case <-d.Context.Done():
		d.killContainer(id, waitCh)
		err = errors.New("Aborted")
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorln("Aborted" )
		d.buildFinish <- err
	case err = <-attachCh:
		d.killContainer(id, waitCh)
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorln("Container", id, "finished with", err)
		d.buildFinish <- err

	case err = <-waitCh:
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorln("Container", id, "finished with", err)
		d.buildFinish <- err
	}
	return nil
}
func (d *DockerExecutor) waitForContainer(ctx context.Context, id string) error {
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Waiting for container", id )

	retries := 0

	// Use active wait
	for ctx.Err() == nil {
		container, err := d.dockerClient.ContainerInspect(ctx, id)
		if err != nil {
			if docker_helper.IsErrNotFound(err) {
				return err
			}
			if retries > 3 {
				return err
			}
			retries++
			time.Sleep(time.Second)
			continue
		}
		// 重置重试计时器
		retries = 0
		if container.State.Running {
			time.Sleep(time.Second)
			continue
		}
		if container.State.ExitCode != 0 {
			return  errors.New(fmt.Sprintf("exit code %d", container.State.ExitCode))
		}
		return nil
	}
	return ctx.Err()
}
func (d *DockerExecutor) Run() error {
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Run")

	var runOn *types.ContainerJSON
	runOn, err := d.requestBuildContainer()
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorf("create Container with error: %v", err)
		d.IsSystemError = true
		return err
	}
	d.runContainer(runOn.ID, bytes.NewBufferString(d.Build.Command))
	return nil
}

func (d *DockerExecutor) getTimeout ()int32 {
	if d.Build.Timeout > 0{
		return d.Build.Timeout
	}
	return common.DefaultLivessTimeout

}

func (d *DockerExecutor) getWorkspace ()string {
	if d.config.Docker.DockerWorkspace != "" {
		return d.config.Docker.DockerWorkspace
	}
	return common.DefaultDockerWorkspace


}
func (d *DockerExecutor) Wait() error {
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Info("Wait")
	// Wait for signals: abort, timeout or finish
	select {
	case <- time.After( time.Duration(d.getTimeout()) * time.Second ):
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Errorln("Build Timeout")
		url := fmt.Sprintf("%s/minio/%s/%s",d.config.MinioConfig.Url,d.config.MinioConfig.Bucket,d.ObjectName)
		kafka.ErrorKafka(d.config.KafkaConfig,d.containerName,d.Build.MetaData,url,errors.New("Build Timeout"),d.IsSystemError)

	case err := <- d.buildFinish:
		// command finished
		if err != nil{
			logrus.WithFields(logrus.Fields{
				"name": d.containerName,
			}).Errorf("build faild with error: %v", err)
			url := fmt.Sprintf("%s/minio/%s/%s",d.config.MinioConfig.Url,d.config.MinioConfig.Bucket,d.ObjectName)
			kafka.ErrorKafka(d.config.KafkaConfig,d.containerName,d.Build.MetaData,url,err,d.IsSystemError)
		} else {
			logrus.WithFields(logrus.Fields{
				"name": d.containerName,
			}).Info("Build succeeded")
			url := fmt.Sprintf("%s/minio/%s/%s",d.config.MinioConfig.Url,d.config.MinioConfig.Bucket,d.ObjectName)
			kafka.SucceededKafka(d.config.KafkaConfig,d.containerName,d.Build.MetaData,url)
		}
	}
	return nil
}

func (d *DockerExecutor)SendError(err error)  {
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
		"error": err,
	}).Infoln("SendError")

	url := fmt.Sprintf("%s/minio/%s/%s",d.config.MinioConfig.Url,d.config.MinioConfig.Bucket,d.ObjectName)
	kafka.ErrorKafka(d.config.KafkaConfig,d.containerName,d.Build.MetaData,url,err,d.IsSystemError)
	d.BuildLog.WriteString(err.Error())
}

func (d *DockerExecutor) cleanBuildLog()  {
	if d.BuildLog != nil {
		if minio.NewMinioHelper(d.config.MinioConfig) != nil{
			_,err := minio.NewMinioHelper(d.config.MinioConfig).PutObject(d.ObjectName,d.BuildLog.Name())
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"name": d.containerName,
				}).Errorf("Minio PutObject with error: %v", err)
			}
			logrus.WithFields(logrus.Fields{
				"name": d.containerName,
				"objectName": d.ObjectName,
			}).Infoln("Minio PutObject ok")
		}
		os.Remove(d.BuildLog.Name())
	}
}

func (d *DockerExecutor) Cleanup() error {
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Info("Cleanup")
	d.cleanBuildLog()

   	d.removeContainer(d.containerName)

	return nil
}

// 可以不删除文件,由后台GC 删除
//func (d *DockerExecutor) CleanFileSystem() {
//	logrus.WithFields(logrus.Fields{
//		"name": d.containerName,
//	}).Info("CleanFileSystem")
//	err := os.RemoveAll(d.CI_PROJECT_DIR.Value)
//	if err != nil {
//		logrus.WithFields(logrus.Fields{
//			"name": d.containerName,
//		}).Info("CleanFileSystem with err ",err)
//	}
//	logrus.WithFields(logrus.Fields{
//		"name": d.containerName,
//	}).Info("CleanFileSystem ok")
//}

func (d *DockerExecutor) killContainer(id string, waitCh chan error) (err error) {
	for {
		logrus.WithFields(logrus.Fields{
			"name": d.containerName,
		}).Infoln("Killing container", id, "...")
		d.dockerClient.ContainerKill(d.Context, id, "SIGKILL")

		// Wait for signal that container were killed
		// or retry after some time
		select {
		case err = <-waitCh:
			return

		case <-time.After(time.Second):
		}
	}
}

func (d *DockerExecutor) removeContainer( id string) error {
	options := types.ContainerRemoveOptions{
		RemoveVolumes: true,
		Force:         true,
	}
	err := d.dockerClient.ContainerRemove(d.Context, id, options)
	logrus.WithFields(logrus.Fields{
		"name": d.containerName,
	}).Infoln("Removed container", id, "with", err)
	return err
}


func createFn() common.Executor {
	return &DockerExecutor{}
}
func init()  {
	common.RegisterExecutor("docker",executors.DefaultExecutorProvider{Creator: createFn})
}